package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import io.swagger.models.auth.In;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;

@Configuration
@Slf4j
public class KafkaStreamHelloListener {
    /**
     * kafka stream 处理器
     * 从上游接受消息   每10秒聚合  发送下游
     *
     * @param streamsBuilder
     * @return
     */
    @Bean // 让spring管理KStream
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //创建kstream对象，同时指定从那个topic中接收消息   {"articleId":"1632279090130194433", "type": "VIEWS"  add:1}
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);// 订阅的主题名词  要和 上游发送消息的主题名称相同


//        stream.flatMapValues(value -> {
//            //空格解析
////            String[] s = value.split(" ");
////            return Arrays.asList(s); //消息中单词集合
//        }).groupBy((key,value) -> value)
//            .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
//            .count()
//            .toStream()
//            .map((key,value) -> new KeyValue<>(key.key(),String.valueOf(value)))

        stream.map((key, value) -> {//每收到一个消息会执行一次
            //key:null   value:{articleId:xxxxxx, add:1,  type:LIKES}
            log.info("处理器收到消息 : key:{}   value:{}", key, value);
            //将value 字符串 转为对象
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            //keyvalue对象中   key为文章id   value为  类型：add的值   如果是点赞value LIKES:1
            return new KeyValue<>(mess.getArticleId().toString(), mess.getType() + ":" + mess.getAdd());
        }).groupByKey()//按照 keyValue中的 key文章id聚合
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))//时间窗口设置为10s
//                .count()//统计次数   意思是统计10秒之内 文章的行为次数
                .aggregate(new Initializer<String>() {//聚合方法
                               @Override
                               public String apply() {
                                   //每个时间窗口 开始时 会执行  可以用来初始化一些  本次时间窗口内的 默认数据
                                   log.info("聚合之前初始化方法执行");
                                   ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
                                   return JSON.toJSONString(mess);//{articleId:null, view:0, collelct:0, comment:0,  like:0}
                               }
                           }, new Aggregator<String, String, String>() {
                               /**
                                *  聚合逻辑    10秒每收到一个消息 就会执行一次apply方法
                                *
                                * @param key 指的时KeyValue对象中的键   文章id
                                * @param value  指的时KeyValue对象中的值  LIKES:1    VIEWS:1
                                * @param aggregate   10秒之内 上一次聚合的结果(注意一个时间窗口内接受第一个消息时，articleId为null)
                                *                   {articleId:文章id, view:0, collelct:0, comment:0,  like:1}
                                * @return
                                */
                               @Override
                               public String apply(String key, String value, String aggregate) {
                                   log.info("聚合逻辑  key:{}  value:{}  aggregate:{}", key, value, aggregate);
                                   ArticleVisitStreamMess mess = JSON.parseObject(aggregate, ArticleVisitStreamMess.class);
                                   if (mess.getArticleId() == null) {
                                       //说明时本时间窗口内第一次收到消息
                                       mess.setArticleId(Long.valueOf(key));
                                   }
                                   String[] split = value.split(":");//["LIKES", "1"]   或者  ["VIEWS", 1]
                                  /* if (split[0].equals("LIKES")) {
                                       mess.setLike(mess.getLike() + Integer.valueOf(split[1]));
                                   }
                                   if (split[0].equals("VIEWS")) {
                                       mess.setView(mess.getView() + Integer.valueOf(split[1]));
                                   }*/

                                   switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                       case COLLECTION:
                                           mess.setCollect(mess.getCollect() + Integer.valueOf(split[1]));
                                           break;
                                       case COMMENT:
                                           mess.setComment(mess.getComment() + Integer.valueOf(split[1]));
                                           break;
                                       case LIKES:
                                           mess.setLike(mess.getLike() + Integer.valueOf(split[1]));
                                           break;
                                       case VIEWS:
                                           mess.setView(mess.getView() + Integer.valueOf(split[1]));
                                           break;
                                   }
                                   log.info("本次消息聚合之后 累加的结果为: {}",JSON.toJSONString(mess));
                                   return JSON.toJSONString(mess);//每次收到消息累加之后的结果
                               }
                           },
                            Materialized.as("aaa")//聚合表示
                )//聚合
                .toStream()
                .map((key, value) -> new KeyValue<>(key.key().toString(), value))//keyvalue对象中key为文章的id，value就是时间窗口结束时  聚合的结果
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);//聚合之后 将消息发送到的主题名称  要和消费者消费的主题保持一致

        return stream;
    }
}